package belf.migrate.engine.job.structure.helper;

import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.exception.DDLException;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.Catalog;
import belf.migrate.api.table.schema.TablePath;
import belf.migrate.api.table.schema.UserDefSeq;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.type.DbObjectType;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.util.PostLog;
import belf.migrate.core.exception.BelfRuntimeException;
import belf.migrate.engine.ddl.BaseDDLConvert;
import belf.migrate.engine.ddl.seq.SeqConvertFactory;
import belf.migrate.api.job.Job;
import belf.migrate.engine.job.JobResult;
import belf.migrate.engine.job.JobResultItem;
import belf.migrate.api.taskconf.Task;
import belf.migrate.api.util.ExceptionUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.time.DateFormatUtils;

import java.util.Date;
import java.util.List;

import static belf.migrate.api.job.JobCode.*;

@Slf4j
public class CreateSeqTool extends NonTableTool {
    public CreateSeqTool(Job job) {
        super(job);
    }

    public CreateSeqTool(Job job, JobContext jobContext) {
        super(job, jobContext);
    }

    @Override
    public JobResult run() {
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "创建序列开始！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        JobResult jobResult = new JobResult(SUCCESS, "创建选中的序列成功");
        Catalog sourceCatalog = jobContext.getSourceCatalog();
        Catalog sinkCatalog = jobContext.getSinkCatalog();
        String sourceSchema = jobContext.getTaskConf().getSourceConf().getSchemaName();

        QualifiersModel qualifiersModel = getSinkQualifiersModel();

        List<String> seqNames = job.getMissions(); //用户是否指定了要创建的序列的名字？
        try {
            List<UserDefSeq> userDefSeqs = sourceCatalog.listSequences(sourceSchema, seqNames);
            if (null != userDefSeqs) {
                for (UserDefSeq userDefSeq : userDefSeqs) {
                    String startTime = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");
                    jobContext.addProgress(DbObjectType.SEQUENCE.getTypeName(), userDefSeqs.size());
                    JobResultItem jobResultItem = process(userDefSeq, sinkCatalog, startTime, qualifiersModel);
                    jobResult.addItem(jobResultItem);
                    if (jobResultItem.getCode() != SUCCESS
                            && job.getTask().getTaskPolicy() == Task.TaskPolicy.ERROR_STOP) {
                        break;  //遇到错误即返回
                    }
                }
            }
        } catch (CatalogException e) {
            e.printStackTrace();
        }
        if (jobResult.getFailtures() > 0) {
            jobResult.setCode(CREATE_SEQ_ERROR);
            jobResult.setMessage("创建选中的序列失败");
            log.error("Create Some Sequences Of {} Failed.", jobContext.getValidSinkDatabase());
        } else {
            log.info("Create All Sequences Of {} Successfully.", jobContext.getValidSinkDatabase());
        }

        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "创建序列结束！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        return jobResult;
    }

    private JobResultItem process(UserDefSeq userDefSeq, Catalog sinkCatalog, String startTime, QualifiersModel qualifiersModel) {
        TablePath tablePath = TablePath.of(jobContext.getTaskConf().getSinkConf().getDatabaseName(),
                jobContext.getTaskConf().getSinkConf().getSchemaName(),
                userDefSeq.getSeqName(),
                qualifiersModel);

        JobResultItem jobResultItem = new JobResultItem(userDefSeq.getSeqName(), tablePath.getCaseTableName(), SUCCESS, "创建序列 " + userDefSeq.getSeqName() + "成功");
        jobResultItem.setType(DbObjectType.SEQUENCE.getTypeName());
        jobResultItem.setStartTime(startTime);

        BaseDDLConvert convert = new SeqConvertFactory().create(jobContext.getValidSinkDatabase(),
                userDefSeq.getSeqName(),
                userDefSeq.getDdl(),
                jobContext.getSourceCatalog(),
                sinkCatalog,
                getSinkQualifiersModel());
        if (null == convert) {
            String endTime = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");

            PostLog.sendLogModel(
                    job.getJobId(),
                    job.getTask().getTaskType().getName(),
                    String.format("缺少%s到%s的序列转换器",
                            jobContext.getTaskConf().getSourceConf().getCatalogName(),
                            jobContext.getTaskConf().getSinkConf().getCatalogName()),
                    LogLevelType.ERROR.getType(),
                    EXECUTING,
                    jobContext.getProgress());

            return new JobResultItem(userDefSeq.getSeqName(),
                    tablePath.getCaseTableName(),
                    CREATE_SEQ_ERROR,
                    String.format("缺少%s到%s的序列转换器",
                            jobContext.getTaskConf().getSourceConf().getCatalogName(),
                            jobContext.getTaskConf().getSinkConf().getCatalogName()),
                    "",
                    DbObjectType.SEQUENCE.getTypeName(),
                    startTime,
                    endTime,
                    userDefSeq.getDdl(),
                    null);
        }

        String ddl = convert.convert();
        try {
            sinkCatalog.executeDdl(tablePath, ddl);
            jobResultItem.setSinkDdl(ddl);
            log.info("Sequence create successfully: \n{}", ddl);
        } catch (DDLException e) {
            log.error("Sequence create ddl: \n{}", ddl);
            e.printStackTrace();
            String endTime = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");
            PostLog.sendLogModel(
                    job.getJobId(),
                    job.getTask().getTaskType().getName(),
                    String.format("创建序列 %s 失败", userDefSeq.getSeqName()),
                    LogLevelType.ERROR.getType(),
                    EXECUTING,
                    jobContext.getProgress());
            String errorTips = null;
            if (!CollectionUtils.isEmpty(convert.getWarnings())) {
                errorTips = String.join(" \n ", convert.getWarnings().stream().map(BelfRuntimeException::getMessage).toList());
            }
            return new JobResultItem(userDefSeq.getSeqName(),
                    tablePath.getCaseTableName(),
                    CREATE_SEQ_ERROR,
                    String.format("创建序列 %s 失败, SQL错误：%s", userDefSeq.getSeqName(), e.getCause().getMessage()),
                    ExceptionUtils.toString(e.getCause()),
                    DbObjectType.SEQUENCE.getTypeName(),
                    startTime,
                    endTime,
                    userDefSeq.getDdl(),
                    ddl,
                    errorTips);
        }

        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                String.format("创建序列 %s 成功", userDefSeq.getSeqName()),
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());

        jobResultItem.setEndTime(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"));
        return jobResultItem;
    }
}
